package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.IndexBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.keyBy.KeySelectFunction;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer09;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Properties;

public class IndexQuotMinTask implements ProcessDataInterface {
    @Override
    public void process(DataStream<CleanBean> waterData) {
//        *
//         * 开发步骤：
//         * 1.定义侧边流
//         * 2.数据分组
//         * 3.划分时间窗口
//         * 4.分时数据处理（新建分时窗口函数）
//         * 5.数据分流
//         * 6.数据分流转换
//         * 7.分表存储(写入kafka)
        // 1.定义侧边流
        OutputTag<IndexBean> outputTag = new OutputTag<IndexBean> ("szseTag", TypeInformation.of(IndexBean.class));
        //* 2.数据分组
        SingleOutputStreamOperator<IndexBean> sse = waterData.keyBy(new KeySelectFunction())
                //         * 3.划分时间窗口
                .timeWindow(Time.minutes(1))
                //         * 4.分时数据处理（新建分时窗口函数）
                .apply(new IndexMinWindowFunction())
                //         * 5.数据分流
                .process(new ProcessFunction<IndexBean, IndexBean>() {
                    @Override
                    public void processElement(IndexBean value, Context ctx, Collector<IndexBean> out) throws Exception {
                        if (value.getSource().equals("sse")) {
                            out.collect(value);
                        } else {
                            ctx.output(outputTag, value);
                        }
                    }
                });
         // * 6.数据分流转换
        SingleOutputStreamOperator<String> szseStr = sse.getSideOutput(outputTag).map(new MapFunction<IndexBean, String>() {
            @Override
            public String map(IndexBean value) throws Exception {
                return JSON.toJSONString(value);
            }
        });
        SingleOutputStreamOperator<String> sseStr = sse.map(new MapFunction<IndexBean, String>() {
            @Override
            public String map(IndexBean value) throws Exception {
                return JSON.toJSONString(value);
            }
        });
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", QuotConfig.config.getProperty("bootstrap.servers"));

        FlinkKafkaProducer011<String> flinkKafkaProducerSse = new FlinkKafkaProducer011<String>(QuotConfig.config.getProperty("sse.index.topic"), new SimpleStringSchema(), properties);
        FlinkKafkaProducer011<String> flinkKafkaProducerSzse = new FlinkKafkaProducer011<String>(QuotConfig.config.getProperty("szse.index.topic"), new SimpleStringSchema(), properties);
        sseStr.addSink(flinkKafkaProducerSse);
        szseStr.addSink(flinkKafkaProducerSzse);
    }
}
